package com.lsx143.wordcount.day2;

import com.lsx143.wordcount.bean.WaterSensor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

import java.util.Arrays;
import java.util.List;

public class MinTest {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);
        List<WaterSensor> waterSensors = Arrays.asList(
                new WaterSensor("sensor_1", 1001L, 10),
                new WaterSensor("sensor_1", 1002L, 20),
                new WaterSensor("sensor_1", 1003L, 5),
                new WaterSensor("sensor_1", 1004L, 40),
                new WaterSensor("sensor_2", 1005L, 10),
                new WaterSensor("sensor_2", 1006L, 40)
        );
        DataStreamSource<WaterSensor> dss = env.fromCollection(waterSensors);
        dss
                .process(new ProcessFunction<WaterSensor, Integer>() {
                    int sum = 0;

                    @Override
                    public void processElement(WaterSensor value, Context ctx, Collector<Integer> out) throws Exception {
                        sum += value.getVc();
                        out.collect(sum);
                    }
                })
                .print();
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
